package com.beiding.natgo.server;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;


@Slf4j
public class ServerWorker {

    //端口号
    private Integer port;

    //ws连接
    private SessionHolder sessionHolder;

    //服务通道
    private ServerSocketChannel serverSocketChannel;

    //工作线程
    private Thread coreThread;

    private Queue<Runnable> taskQueue = new ConcurrentLinkedQueue<>();

    //通道map
    private Map<String, ChanelHolder> chanelHolderMap = new HashMap<>();

    //继续标记
    private boolean goFlag = true;

    //选择器
    private Selector selector;

    public ServerWorker(SessionHolder sessionHolder, Integer port) {
        this.port = port;
        this.sessionHolder = sessionHolder;
    }

    public void stop() {
        this.goFlag = false;
        if (selector != null) {
            selector.wakeup();
        }
    }

    public void start() {
        coreThread = new Thread() {
            @Override
            public void run() {
                bind();
            }
        };
        coreThread.start();
    }

    private void bind() {

        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            InetSocketAddress localAddress = new InetSocketAddress(port);
            serverSocketChannel.bind(localAddress);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            sessionHolder.sendInitSuccess();
            handleEvent();
            try {
                log.info("关闭通道,释放端口{}", port);
                selector.close();
                serverSocketChannel.close();
                log.info("通道已关闭");
            } catch (Exception e) {
                e.printStackTrace();
            }
        } catch (Exception e) {
            e.printStackTrace();
            sessionHolder.sendInitFail(e.getMessage());
        }

    }


    private void handleEvent() {
        while (goFlag) {
            try {
                int n = selector.select();

                Runnable poll;
                while ((poll = taskQueue.poll()) != null) {
                    poll.run();
                }

                if (n == 0) {
                    continue;
                }

                Set<SelectionKey> selectionKeys = selector.selectedKeys();

                Iterator<SelectionKey> iterator = selectionKeys.iterator();

                while (iterator.hasNext()) {
                    SelectionKey next = iterator.next();
                    iterator.remove();
                    if (next.isReadable()) {
                        ChanelHolder attachment = (ChanelHolder) next.attachment();
                        attachment.handleRead();
                    } else if (next.isAcceptable()) {
                        SocketChannel channel = serverSocketChannel.accept();
                        channel.configureBlocking(false);

                        //创建通道
                        ChanelHolder chanelHolder = new ChanelHolder(this, channel, sessionHolder);

                        String id = chanelHolder.getId();

                        //记录通道
                        chanelHolderMap.put(id, chanelHolder);

                        //发送创建新连接的消息
                        sessionHolder.sendCreateConnect(id);

                    }
                }

            } catch (IOException e) {
                e.printStackTrace();
                sessionHolder.sendServerError(e.getMessage());
            }
        }
    }

    public void createChanelSuccess(String chanelId) {
        ChanelHolder chanelHolder = this.chanelHolderMap.get(chanelId);
        if (chanelHolder != null) {
            SocketChannel socketChannel = chanelHolder.getSocketChannel();
            offerTask(() -> {
                try {
                    SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_READ, chanelHolder);
                    chanelHolder.setSelectionKey(selectionKey);
                } catch (IOException e) {
                    e.printStackTrace();
                    this.chanelHolderMap.remove(chanelId);
                }
            });

        }
    }

    private void offerTask(Runnable runnable) {
        taskQueue.offer(runnable);
        selector.wakeup();
    }

    public void createChanelFail(String chanelId) {
        ChanelHolder chanelHolder = this.chanelHolderMap.remove(chanelId);
        if (chanelHolder != null) {
            try {
                chanelHolder.getSocketChannel().close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    //自行移除通道 并向客户端发送消息
    public void removeChanelAndSendClose(String id) {
        ChanelHolder chanelHolder = this.chanelHolderMap.remove(id);
        if (chanelHolder != null) {
            sessionHolder.sendCloseChanel(id);
        }
    }

    public void writeData(String chanelId, byte[] data) {
        ChanelHolder chanelHolder = this.chanelHolderMap.get(chanelId);
        if (chanelHolder != null) {
            chanelHolder.writeData(data);
        } else {
            sessionHolder.sendCloseChanel(chanelId);
        }
    }


    public void closeChanel(String chanelId) {
        ChanelHolder chanelHolder = this.chanelHolderMap.remove(chanelId);
        if (chanelHolder != null) {
            SocketChannel socketChannel = chanelHolder.getSocketChannel();
            if (socketChannel != null) {
                try {
                    socketChannel.shutdownInput();
                    socketChannel.shutdownOutput();
                    socketChannel.close();
                    chanelHolder.getSelectionKey().cancel();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
